package com.flink.flinkdemo.demo;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.security.Key;

/**
 * @author zhangy on 2020/12/31
 * @description: add you description
 */
public class StreamJoinDemo {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> sour1 = env.socketTextStream("localhost", 9001);


        DataStreamSource<String> sour2 = env.socketTextStream("localhost", 9002);

        SingleOutputStreamOperator<Tuple2<String, String>> flat1 = sour1.flatMap(new MyFlatMap());

        SingleOutputStreamOperator<Tuple2<String, String>> flat2 = sour2.flatMap(new MyFlatMap());



    }

    public static class MyFlatMap implements FlatMapFunction<String, Tuple2<String, String>>{
        @Override
        public void flatMap(String value, Collector<Tuple2<String, String>> out) throws Exception {
            String[] words = value.split(" ");
            out.collect(new Tuple2<>(words[0], words[1]));
        }
    }

    public static class MyKeySelector implements KeySelector<Tuple2<String,String>,Tuple2<String,String>>{

        @Override
        public Tuple2<String, String> getKey(Tuple2<String, String> value) throws Exception {
            return null;
        }
    }
}